<?php
/**
 * 批量上传任务到 Redis Stream 队列
 */

// 连接 Redis
$redis = new Redis();
$redis->connect('127.0.0.1', 6379);
$redis->select(1);

// 配置
$taskStream = 'task:stream:queue';
$callbackStream = 'task:callback:queue';
$groupName = 'task_processor_group';

echo "=== 批量上传任务到 Stream 队列 ===\n\n";

try {
    // 1. 准备多个任务数据
    $tasks = [
        [
            'task_type' => 'data_process',
            'task_data' => [
                'input' => [1, 2, 3, 4, 5],
                'process_type' => 'sum'
            ],
            'priority' => 'high',
            'callback' => [
                'type' => 'http',
                'url' => 'http://example.com/callback1',
                'method' => 'POST'
            ]
        ],
        [
            'task_type' => 'data_process',
            'task_data' => [
                'input' => ['apple', 'banana', 'cherry'],
                'process_type' => 'transform'
            ],
            'priority' => 'normal',
            'callback' => [
                'type' => 'http',
                'url' => 'http://example.com/callback2',
                'method' => 'POST'
            ]
        ],
        [
            'task_type' => 'file_process',
            'task_data' => [
                'file_path' => '/tmp/test.txt',
                'operation' => 'write',
                'content' => 'Hello World!'
            ],
            'priority' => 'low',
            'callback' => [
                'type' => 'http',
                'url' => 'http://example.com/callback3',
                'method' => 'POST'
            ]
        ],
        [
            'task_type' => 'email_send',
            'task_data' => [
                'to' => 'user@example.com',
                'subject' => '测试邮件',
                'content' => '这是一封测试邮件'
            ],
            'priority' => 'high',
            'callback' => [
                'type' => 'http',
                'url' => 'http://example.com/callback4',
                'method' => 'POST'
            ]
        ],[
            'task_type' => 'email_send',
            'task_data' => [
                'to' => 'user@example.com',
                'subject' => '测试邮件',
                'content' => '这是一封测试邮件'
            ],
            'priority' => 'high',
            'callback' => [
                'type' => 'http',
                'url' => 'http://example.com/callback4',
                'method' => 'POST'
            ]
        ],[
            'task_type' => 'email_send',
            'task_data' => [
                'to' => 'user@example.com',
                'subject' => '测试邮件',
                'content' => '这是一封测试邮件'
            ],
            'priority' => 'high',
            'callback' => [
                'type' => 'http',
                'url' => 'http://example.com/callback4',
                'method' => 'POST'
            ]
        ],[
            'task_type' => 'email_send',
            'task_data' => [
                'to' => 'user@example.com',
                'subject' => '测试邮件',
                'content' => '这是一封测试邮件'
            ],
            'priority' => 'high',
            'callback' => [
                'type' => 'http',
                'url' => 'http://example.com/callback4',
                'method' => 'POST'
            ]
        ],[
            'task_type' => 'email_send',
            'task_data' => [
                'to' => 'user@example.com',
                'subject' => '测试邮件',
                'content' => '这是一封测试邮件'
            ],
            'priority' => 'high',
            'callback' => [
                'type' => 'http',
                'url' => 'http://example.com/callback4',
                'method' => 'POST'
            ]
        ],[
            'task_type' => 'email_send',
            'task_data' => [
                'to' => 'user@example.com',
                'subject' => '测试邮件',
                'content' => '这是一封测试邮件'
            ],
            'priority' => 'high',
            'callback' => [
                'type' => 'http',
                'url' => 'http://example.com/callback4',
                'method' => 'POST'
            ]
        ],[
            'task_type' => 'email_send',
            'task_data' => [
                'to' => 'user@example.com',
                'subject' => '测试邮件',
                'content' => '这是一封测试邮件'
            ],
            'priority' => 'high',
            'callback' => [
                'type' => 'http',
                'url' => 'http://example.com/callback4',
                'method' => 'POST'
            ]
        ],[
            'task_type' => 'email_send',
            'task_data' => [
                'to' => 'user@example.com',
                'subject' => '测试邮件',
                'content' => '这是一封测试邮件'
            ],
            'priority' => 'high',
            'callback' => [
                'type' => 'http',
                'url' => 'http://example.com/callback4',
                'method' => 'POST'
            ]
        ],[
            'task_type' => 'email_send',
            'task_data' => [
                'to' => 'user@example.com',
                'subject' => '测试邮件',
                'content' => '这是一封测试邮件'
            ],
            'priority' => 'high',
            'callback' => [
                'type' => 'http',
                'url' => 'http://example.com/callback4',
                'method' => 'POST'
            ]
        ],[
            'task_type' => 'email_send',
            'task_data' => [
                'to' => 'user@example.com',
                'subject' => '测试邮件',
                'content' => '这是一封测试邮件'
            ],
            'priority' => 'high',
            'callback' => [
                'type' => 'http',
                'url' => 'http://example.com/callback4',
                'method' => 'POST'
            ]
        ],[
            'task_type' => 'email_send',
            'task_data' => [
                'to' => 'user@example.com',
                'subject' => '测试邮件',
                'content' => '这是一封测试邮件'
            ],
            'priority' => 'high',
            'callback' => [
                'type' => 'http',
                'url' => 'http://example.com/callback4',
                'method' => 'POST'
            ]
        ],[
            'task_type' => 'email_send',
            'task_data' => [
                'to' => 'user@example.com',
                'subject' => '测试邮件',
                'content' => '这是一封测试邮件'
            ],
            'priority' => 'high',
            'callback' => [
                'type' => 'http',
                'url' => 'http://example.com/callback4',
                'method' => 'POST'
            ]
        ],[
            'task_type' => 'email_send',
            'task_data' => [
                'to' => 'user@example.com',
                'subject' => '测试邮件',
                'content' => '这是一封测试邮件'
            ],
            'priority' => 'high',
            'callback' => [
                'type' => 'http',
                'url' => 'http://example.com/callback4',
                'method' => 'POST'
            ]
        ],[
            'task_type' => 'email_send',
            'task_data' => [
                'to' => 'user@example.com',
                'subject' => '测试邮件',
                'content' => '这是一封测试邮件'
            ],
            'priority' => 'high',
            'callback' => [
                'type' => 'http',
                'url' => 'http://example.com/callback4',
                'method' => 'POST'
            ]
        ],[
            'task_type' => 'email_send',
            'task_data' => [
                'to' => 'user@example.com',
                'subject' => '测试邮件',
                'content' => '这是一封测试邮件'
            ],
            'priority' => 'high',
            'callback' => [
                'type' => 'http',
                'url' => 'http://example.com/callback4',
                'method' => 'POST'
            ]
        ],[
            'task_type' => 'email_send',
            'task_data' => [
                'to' => 'user@example.com',
                'subject' => '测试邮件',
                'content' => '这是一封测试邮件'
            ],
            'priority' => 'high',
            'callback' => [
                'type' => 'http',
                'url' => 'http://example.com/callback4',
                'method' => 'POST'
            ]
        ],[
            'task_type' => 'email_send',
            'task_data' => [
                'to' => 'user@example.com',
                'subject' => '测试邮件',
                'content' => '这是一封测试邮件'
            ],
            'priority' => 'high',
            'callback' => [
                'type' => 'http',
                'url' => 'http://example.com/callback4',
                'method' => 'POST'
            ]
        ],[
            'task_type' => 'email_send',
            'task_data' => [
                'to' => 'user@example.com',
                'subject' => '测试邮件',
                'content' => '这是一封测试邮件'
            ],
            'priority' => 'high',
            'callback' => [
                'type' => 'http',
                'url' => 'http://example.com/callback4',
                'method' => 'POST'
            ]
        ],[
            'task_type' => 'email_send',
            'task_data' => [
                'to' => 'user@example.com',
                'subject' => '测试邮件',
                'content' => '这是一封测试邮件'
            ],
            'priority' => 'high',
            'callback' => [
                'type' => 'http',
                'url' => 'http://example.com/callback4',
                'method' => 'POST'
            ]
        ],[
            'task_type' => 'email_send',
            'task_data' => [
                'to' => 'user@example.com',
                'subject' => '测试邮件',
                'content' => '这是一封测试邮件'
            ],
            'priority' => 'high',
            'callback' => [
                'type' => 'http',
                'url' => 'http://example.com/callback4',
                'method' => 'POST'
            ]
        ],
        // [
        //     'task_type' => 'data_validate',
        //     'task_data' => [
        //         'input' => ['data1', 'data2', 'data3'],
        //         'rules' => ['required', 'email', 'numeric']
        //     ],
        //     'priority' => 'normal',
        //     'callback' => [
        //         'type' => 'http',
        //         'url' => 'http://example.com/callback5',
        //         'method' => 'POST'
        //     ]
        // ]
    ];
    
    echo "1. 准备批量任务数据\n";
    echo "   任务数量: " . count($tasks) . "\n\n";
    
    // 2. 确保消费者组存在
    try {
        $redis->xGroup('CREATE', $taskStream, $groupName, '0', true);
        echo "2. 消费者组创建成功\n";
        echo "   组名: {$groupName}\n\n";
    } catch (RedisException $e) {
        if (strpos($e->getMessage(), 'BUSYGROUP') !== false) {
            echo "2. 消费者组已存在\n";
            echo "   组名: {$groupName}\n\n";
        } else {
            throw $e;
        }
    }
    
    // 3. 批量上传任务
    echo "3. 开始批量上传任务\n";
    $uploadedTasks = [];
    $successCount = 0;
    $failedCount = 0;
    
    foreach ($tasks as $index => $task) {
        try {
            $taskId = uniqid('task_', true);
            $taskData = [
                'task_id' => $taskId,
                'task_type' => $task['task_type'],
                'task_data' => json_encode($task['task_data'], JSON_UNESCAPED_UNICODE),
                'created_at' => time(),
                'status' => 'pending',
                'priority' => $task['priority'],
                'retry_count' => 0,
                'max_retries' => 3,
                'timeout' => 300,
                'callback' => json_encode($task['callback'], JSON_UNESCAPED_UNICODE)
            ];
            
            $messageId = $redis->xAdd($taskStream, '*', $taskData);
            
            $uploadedTasks[] = [
                'index' => $index + 1,
                'task_id' => $taskId,
                'message_id' => $messageId,
                'task_type' => $task['task_type'],
                'priority' => $task['priority'],
                'success' => true
            ];
            
            $successCount++;
            
            echo "   ✓ 任务 " . ($index + 1) . " 上传成功\n";
            echo "     任务ID: {$taskId}\n";
            echo "     消息ID: {$messageId}\n";
            echo "     类型: {$task['task_type']}\n";
            echo "     优先级: {$task['priority']}\n\n";
            
        } catch (Exception $e) {
            $uploadedTasks[] = [
                'index' => $index + 1,
                'task_type' => $task['task_type'],
                'error' => $e->getMessage(),
                'success' => false
            ];
            
            $failedCount++;
            
            echo "   ✗ 任务 " . ($index + 1) . " 上传失败\n";
            echo "     类型: {$task['task_type']}\n";
            echo "     错误: {$e->getMessage()}\n\n";
        }
    }
    
    // 4. 显示上传结果
    echo "4. 批量上传结果\n";
    echo "   总数: " . count($tasks) . "\n";
    echo "   成功: {$successCount}\n";
    echo "   失败: {$failedCount}\n\n";
    
    // 5. 检查 Stream 信息
    echo "5. Stream 信息\n";
    $streamInfo = $redis->xInfo('STREAM', $taskStream);
    echo "   长度: {$streamInfo['length']}\n";
    echo "   最后生成ID: {$streamInfo['last-generated-id']}\n\n";
    
    // 6. 检查消费者组信息
    echo "6. 消费者组信息\n";
    $groupInfo = $redis->xInfo('GROUPS', $taskStream);
    foreach ($groupInfo as $group) {
        echo "   组名: {$group['name']}\n";
        echo "   消费者数量: {$group['consumers']}\n";
        echo "   待确认消息: {$group['pending']}\n";
        echo "   最后投递ID: {$group['last-delivered-id']}\n\n";
    }
    
    // 7. 显示最近的消息
    echo "7. 最近的消息\n";
    $messages = $redis->xRead(['streams' => [$taskStream => 0]], 10, 1000);
    if (!empty($messages[$taskStream])) {
        $count = 0;
        foreach ($messages[$taskStream] as $msgId => $fields) {
            $count++;
            echo "   消息 {$count}:\n";
            echo "     消息ID: {$msgId}\n";
            echo "     任务ID: " . ($fields['task_id'] ?? 'N/A') . "\n";
            echo "     任务类型: " . ($fields['task_type'] ?? 'N/A') . "\n";
            echo "     优先级: " . ($fields['priority'] ?? 'N/A') . "\n";
            echo "     状态: " . ($fields['status'] ?? 'N/A') . "\n\n";
        }
    } else {
        echo "   暂无消息\n\n";
    }
    
    // 8. 列出所有相关的 keys
    echo "8. 相关 Redis Keys\n";
    $keys = $redis->keys('task:*');
    foreach ($keys as $key) {
        $type = $redis->type($key);
        echo "   {$key} ({$type})\n";
    }
    echo "\n";
    
    echo "=== 批量上传完成 ===\n";
    echo "成功上传: {$successCount} 个任务\n";
    echo "失败: {$failedCount} 个任务\n";
    echo "Stream: {$taskStream}\n";
    echo "状态: 任务已添加到队列，等待处理\n";
    
    // 9. 显示上传的任务列表
    echo "\n上传的任务列表:\n";
    foreach ($uploadedTasks as $task) {
        if ($task['success']) {
            echo "✓ 任务 {$task['index']}: {$task['task_type']} (ID: {$task['task_id']})\n";
        } else {
            echo "✗ 任务 {$task['index']}: {$task['task_type']} (错误: {$task['error']})\n";
        }
    }
    
} catch (Exception $e) {
    echo "错误: " . $e->getMessage() . "\n";
    echo "堆栈跟踪: " . $e->getTraceAsString() . "\n";
} 